package com.hex.ds.hdrs.engine.manager;

import cn.hutool.core.util.StrUtil;
import com.hex.ds.hdrs.common.constant.Const;
import com.hex.ds.hdrs.common.helper.ListHelper;
import com.hex.ds.hdrs.period.job.po.JobTask;
import com.hex.ds.hdrs.period.task.po.PeriodTask;
import com.hex.ds.hdrs.period.task.util.TaskTrackHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @Package: com.hex.ds.hdrs.engine.manager
 * @ClassName ToDoManager
 * @Description: 待处理队列管理
 * @Author: jd.chen
 * @Date 2021/7/5 17:55
 * @Version v1.0
 **/
@Slf4j
@Component
public class ToDoManager implements ApplicationContextAware {

    private static ApplicationContext context;

    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock readLock = rwl.readLock();
    private final ReentrantReadWriteLock.WriteLock writeLock = rwl.writeLock();


    @Value("${processor.sleep}")
    private int processorSleep;
    @Value("${processor.pool-size}")
    private int poolSize;
    @Value("${logging.showMemoryList}")
    private boolean showMemoryList;

    /**
     * toDoJobList 同步锁
     */
    private final byte[] listSyn = {};

    /**
     * 打标
     */
    private volatile String mark;
    /**
     * 上次查询时间
     */
    private String lastQryTs;
    /**
     * 本次查询时间
     */
    private String curQryTs;
    /**
     * 待处理作业list
     */
    private volatile List<JobTask> toDoJobList = Collections.synchronizedList(new ArrayList<>());
    /**
     * 待处理作业map
     */
    private volatile Map<String, JobTask> toDoJobMap = new ConcurrentHashMap<>();
    /**
     * 待处理作业queue
     */
    private volatile BlockingQueue<JobTask> toDoQueue = new LinkedBlockingQueue<>();
    /**
     * 全量加载时，触发下游作业任务实例，不满足上游依赖先存到delta队列中，等待全量结束DB数据的合并后再合并delta队列中的数据
     */
    private List<JobTask> deltaAppJobList = Collections.synchronizedList(new ArrayList<>());
    /**
     * 全量加载时，处理器、任务执行完成触发下游任务等情况下，作业任务实例满足依赖条件的添加到可执行队列中，
     * 同时将该作业任务实例在deltaDelJobMap中存一份，处理器在处理作业任务时过滤这些作业任务实例
     */
    private Map<String, JobTask> deltaDelJobMap = new ConcurrentHashMap<>();

    /*
     * @Method: init <br>
     * @Param: [jobTaskList] <br>
     * @Return: void <br>
     * @Description：初始化待处理队列<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 17:24 <br>
     * @Version： V2.0.2<br>
     */
    public void init(List<JobTask> jobTaskList) {
        addJobList(jobTaskList);
        this.lastQryTs = this.curQryTs;
    }

    /*
     * @Method: clear <br>
     * @Param: [] <br>
     * @Return: void <br>
     * @Description：清空待处理队列中的作业/任务实例数据<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:15 <br>
     * @Version： V2.0.2<br>
     */
    public void clear() {
        synchronized (listSyn) {
            this.toDoJobList.clear();
            this.toDoJobMap.clear();
            this.toDoQueue.clear();
        }
    }

    public BlockingQueue<JobTask> getToDoQueue() {
        return toDoQueue;
    }

    /*
     * @Method: getJob <br>
     * @Param: [jobCode] <br>
     * @Return: com.hex.ds.hdrs.period.job.po.JobTask <br>
     * @Description：根据作业编码获取待处理队列中的作业任务实例对象<br>
     *  ConcurrentHashMap.get() 方法时线程安全的
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:36 <br>
     * @Version： V2.0.2<br>
     */
    public JobTask getJob(String jobCode) {
        return toDoJobMap.get(jobCode);
    }

    /*
     * @Method: getDeltaDelJobMap <br>
     * @Param: [jobCode] <br>
     * @Return: com.hex.ds.hdrs.period.job.po.JobTask <br>
     * @Description：根据业务编码获取DeltaMap中作业任务实例对象<br>
     * ConcurrentHashMap.get() 方法时线程安全的
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:36 <br>
     * @Version： V2.0.2<br>
     */
    public JobTask getDeltaDelJobMap(String jobCode) {
        return deltaDelJobMap.get(jobCode);
    }

    /*
     * @Method: removeDelMapJob <br>
     * @Param: [job] <br>
     * @Return: void <br>
     * @Description：DeltaMap删除指定的作业任务实例<br>
     * ConcurrentHashMap.remove() 方法时线程安全的
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:45 <br>
     * @Version： V2.0.2<br>
     */
    public void removeDelMapJob(JobTask job) {
        this.deltaDelJobMap.remove(job.getJobCode());
    }

    /*
     * @Method: removeJob <br>
     * @Param: [job] <br>
     * @Return: void <br>
     * @Description：删除待处理队列中指定的作业任务实例<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:46 <br>
     * @Version： V2.0.2<br>
     */
    public void removeJob(JobTask job) {
        log.debug("【ToDoManager.removeJob】删除指定的作业任务实例，作业编码:{}, 业务日期:{}", job.getJobCode(), job.getBizDate());
        synchronized (listSyn) {
            this.toDoJobList.remove(job);
            this.toDoJobMap.remove(job.getJobCode());
            this.toDoQueue.remove(job);
        }
    }

    /*
     * @Method: poll <br>
     * @Param: [] <br>
     * @Return: com.hex.ds.hdrs.period.job.po.JobTask <br>
     * @Description：消费待处理队列todoQueue中头部作业任务实例对象<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:48 <br>
     * @Version： V2.0.2<br>
     */
    public JobTask poll() {
        try {
            JobTask queueJobTask = null;
            synchronized (this) {
                queueJobTask = toDoQueue.poll();
            }
            if (queueJobTask == null) {
                log.debug("【toDo poll】队列中无任务======");
                return null;
            }
            String jobCode = queueJobTask.getJobCode();
            String queueBizDate = queueJobTask.getBizDate();
            JobTask jobTask = this.getJob(jobCode);
            if (jobTask != null) {
                String mapBizDate = jobTask.getBizDate();
                if (mapBizDate.equals(queueBizDate)) {
                    return queueJobTask;
                } else {
                    log.warn("业务日期异常【toDo poll】job【{}】、queueBizDate【{}】、mapBizDate【{}】", jobCode, queueBizDate, mapBizDate);
                    return null;
                }
            } else {
                log.warn("queue存在作业，map中不存在【toDo poll】job【{}】【{}】", jobCode, queueBizDate);
                return null;
            }
        } catch (Exception e) {
            log.error("【toDo poll】异常:{}", e.getMessage(), e);
            return null;
        }
    }

    /*
     * @Method: addJob <br>
     * @Param: [jobTask, type] <br>
     * @Return: void <br>
     * @Description：向可待处理队列中添加作业任务实例（支持全量加载情况）<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:52 <br>
     * @Version： V2.0.2<br>
     */
    public void addJob(JobTask jobTask, String type) {
        if (Const.ALL_LOADER_MARK.equals(mark)) {
            addDeltaListJob(jobTask);
        } else {
            addToDoJob(jobTask, type);
        }
    }

    /*
     * @Method: addDeltaListJob <br>
     * @Param: [jobTask] <br>
     * @Return: void <br>
     * @Description：全量加载时，将作业任务实例添加到deltaList中，等待DB数据处理完后再进行合并<br>
     * Collections.synchronizedList(new ArrayList<>()) 的add()方法线程安全
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:52 <br>
     * @Version： V2.0.2<br>
     */
    private void addDeltaListJob(JobTask jobTask) {
        log.debug("【ToDoManager.addDeltaListJob】全量加载时，先将作业任务实例添加到DeltaList中，作业编码:{},业务日期:{}", jobTask.getJobCode(), jobTask.getBizDate());
        this.deltaAppJobList.add(jobTask);
    }

    /*
     * @Method: addToDoJob <br>
     * @Param: [jobTask, type] <br>
     * @Return: void <br>
     * @Description：向可待处理队列中添加作业任务实例(不支持全量情况)<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 16:53 <br>
     * @Version： V2.0.2<br>
     */
    public void addToDoJob(JobTask jobTask, String type) {
        log.debug("【ToDoManager.addToDoJob】待处理队列添加新的作业任务实例，作业编码:{},业务日期:{}", jobTask.getJobCode(), jobTask.getBizDate());
        TaskTrackHelper.addTaskTrack(jobTask.getCurTask(), TaskTrackHelper.loadSucc(type));
        synchronized (listSyn) {
            if (!isContainsJob(jobTask.getJobCode())) {
                this.toDoJobList.add(jobTask);
                this.toDoJobMap.put(jobTask.getJobCode(), jobTask);
                this.addJobToQueue(jobTask);
            }
        }
    }

    /*
     * @Method: addJobList <br>
     * @Param: [jobList] <br>
     * @Return: void <br>
     * @Description：重置待处理队列<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 17:22 <br>
     * @Version： V2.0.2<br>
     */
    public void addJobList(List<JobTask> jobList) {
        clear();
        batchAddJobList(jobList);
    }

    /*
     * @Method: batchAddJobList <br>
     * @Param: [jobList] <br>
     * @Return: void <br>
     * @Description：批量向待处理队列中添加作业任务实例<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 17:20 <br>
     * @Version： V2.0.2<br>
     */
    private void batchAddJobList(List<JobTask> jobList) {
        log.info("【ToDoManager.batchAddJobList】批量向待处理队列中添加作业任务实例，添加个数:{}", jobList.size());
        synchronized (listSyn) {
            this.toDoJobList.addAll(jobList);
            this.toDoJobMap.putAll(ListHelper.jobTaskListToMap(jobList));
            this.toDoQueue = ListHelper.listToQueue(jobList);
        }
    }

    /*
     * @Method: clearDeltaAppJobList <br>
     * @Param: [] <br>
     * @Return: void <br>
     * @Description：清空DeltaList中数据<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:16 <br>
     * @Version： V2.0.2<br>
     */
    public void clearDeltaAppJobList() {
        this.deltaAppJobList.clear();
    }

    /**
     * 任务添加到queue
     *
     * @param job
     */
    public void addJobToQueue(JobTask job) {
        // TODO 是否该使用 LinkedBlockingQueue的put方法
        log.debug("+++++++++【toDo addJobToQueue】作业【{}】、日期【{}】", job.getJobCode(), job.getBizDate());
        if (this.toDoQueue.contains(job)) {
            log.debug("【toDo addJobToQueue】队列中已存在作业【{}】、日期【{}】", job.getJobCode(), job.getBizDate());
        } else {
            this.toDoQueue.add(job);
        }
    }

    /*
     * @Method: toDoToToRun <br>
     * @Param: [job, task, type, toRunManager] <br>
     * @Return: void <br>
     * @Description：待处理任务队列到可执行任务队列<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:34 <br>
     * @Version： V2.0.2<br>
     */
    // TODO 该方法不应该放到这里
    public void toDoToToRun(JobTask job, PeriodTask task, String type, ToRunManager toRunManager) {
        log.info("【ToDoManager.toDoToToRun】待执行队列中的作业/任务实例转移到可执行队列中，作业编码: {}，业务日期: {}", job.getJobCode(), job.getBizDate());
        if (Const.ALL_LOADER_MARK.equals(getMark())) {
            addJobToDeltaMap(job);
        }
        toRunManager.addJob(job, task, type);
        removeJob(job);
    }

    /*
     * @Method: addJobToDeltaMap <br>
     * @Param: [jobTask] <br>
     * @Return: void <br>
     * @Description：向DeltaMap中添加作业/任务实例<br>
     *  ConcurrentHashMap.put()方法线程安全
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:23 <br>
     * @Version： V2.0.2<br>
     */
    public void addJobToDeltaMap(JobTask jobTask){
        log.debug("【ToDoManager.addJobToDeltaMap】向DeltaMap中添加作业/任务实例,作业编码: {}, 业务日期: {}" + jobTask.getJobCode(), jobTask.getBizDate());
        deltaDelJobMap.put(jobTask.getJobCode(), jobTask);
    }

    /*
     * @Method: threadSleep <br>
     * @Param: [] <br>
     * @Return: void <br>
     * @Description：处理器休眠<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:17 <br>
     * @Version： V2.0.2<br>
     */
    public void threadSleep() {
        try {
            Thread.sleep(processorSleep);
        } catch (InterruptedException e) {
            log.error("【ToDoManager.threadSleep】休眠异常:{}", e.getMessage(), e);
            Thread.currentThread().interrupt();
        }
    }

    /*
     * @Method: isContainsJob <br>
     * @Param: [jobCode] <br>
     * @Return: boolean <br>
     * @Description：判断指定作业编码是否在待处理队列中<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:39 <br>
     * @Version： V2.0.2<br>
     */
    public boolean isContainsJob(String jobCode) {
        return this.getJob(jobCode) != null
                && this.toDoJobList.contains(this.getJob(jobCode))
                && this.toDoQueue.contains(this.getJob(jobCode));
    }

    /*
     * @Method: isContainsDelJob <br>
     * @Param: [jobCode] <br>
     * @Return: boolean <br>
     * @Description：判断指定作业/任务实例是否在DeltaMap中<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:38 <br>
     * @Version： V2.0.2<br>
     */
    public boolean isContainsDelJob(String jobCode) {
        return this.getDeltaDelJobMap(jobCode) != null;
    }

    /*
     * @Method: setMark <br>
     * @Param: [mark] <br>
     * @Return: void <br>
     * @Description：更新处理器打标状态<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/21 18:19 <br>
     * @Version： V2.0.2<br>
     */
    public void setMark(String mark) {
        writeLock.lock();
        try {
            this.mark = mark;
        } finally {
            writeLock.unlock();
        }
    }

    public String getLastQryTs() {
        return lastQryTs;
    }

    public List<JobTask> getDeltaAppJobList() {
        return deltaAppJobList;
    }

    public int getPoolSize() {
        return poolSize;
    }

    public List<JobTask> getToDoJobList() {
        return toDoJobList;
    }

    public String getMark() {
        readLock.lock();
        try {
            return mark;
        } finally {
            readLock.unlock();
        }
    }

    public boolean loaderMark(String loadMark) {
        boolean flag = false;
        readLock.lock();
        if (StrUtil.equalsAnyIgnoreCase(Const.FREE_LOADER_MARK, mark)) {
            readLock.unlock();
            writeLock.lock();
            try {
                if (StrUtil.equalsAnyIgnoreCase(Const.FREE_LOADER_MARK, mark)) {
                    this.mark = loadMark;
                    flag = true;
                }
                readLock.lock();
            } finally {
                writeLock.unlock();
            }
        }
        try {
            return flag;
        } finally {
            readLock.unlock();
        }
    }

    public String getCurQryTs() {
        return curQryTs;
    }

    public void setCurQryTs(String qryTs) {
        this.curQryTs = qryTs;
    }

    public void setLastQryTs(String lastQryTs) {
        this.lastQryTs = lastQryTs;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        context = applicationContext;
    }
    public static ToDoManager getInstance() {
        return context.getBean(ToDoManager.class);
    }
}
